package com.shujia.flink.sink

import java.sql.{Connection, DriverManager, PreparedStatement}

import com.shujia.flink.source.Student
import org.apache.flink.configuration.Configuration
import org.apache.flink.connector.jdbc.{JdbcConnectionOptions, JdbcSink, JdbcStatementBuilder}
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

object Demo2MysqlSink {
  def main(args: Array[String]): Unit = {


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    val ds1: DataStream[String] = env.readTextFile("flink/data/students.txt")


    //使用自定义sink
    //ds1.addSink(new MysqlSink)

    //使用官网的jdbc
    ds1.addSink(JdbcSink.sink(
      "insert into student(id,name,age,gender,clazz) values(?,?,?,?,?)"
      , new JdbcStatementBuilder[String] {
        override def accept(stat: PreparedStatement, value: String): Unit = {
          val split: Array[String] = value.split(",")
          val id: Int = split(0).toInt
          val name: String = split(1)
          val age: Int = split(2).toInt
          val gender: String = split(3)
          val clazz: String = split(4)

          stat.setInt(1, id)
          stat.setString(2, name)
          stat.setInt(3, age)
          stat.setString(4, gender)
          stat.setString(5, clazz)
        }
      }
      ,
      new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
        .withUrl("jdbc:mysql://master:3306/student?useUnicode=true&characterEncoding=utf-8&useSSL=false")
        .withDriverName("com.mysql.jdbc.Driver")
        .withUsername("root")
        .withPassword("123456")
        .build()
    ))


    env.execute()

  }

}

/**
  * SinkFunction
  * RichSinkFunction  多了open和close 方法用于打开和关闭链接
  *
  */

class MysqlSink extends RichSinkFunction[String] {

  var con: Connection = _

  /**
    * open:方法每一个线程中执行一次，一般用于打开链接
    * 在invoke之前执行
    *
    */

  override def open(parameters: Configuration): Unit = {
    println("创建链接")
    //创建链接
    Class.forName("com.mysql.jdbc.Driver")
    con = DriverManager.getConnection("jdbc:mysql://master:3306/student?useUnicode=true&characterEncoding=utf-8&useSSL=false", "root", "123456")

  }


  //在invoke 之后执行
  override def close(): Unit = {
    println("关闭链接")
    con.close()
  }

  /**
    * invoke: 方法，每一条数据都会执行一次
    *
    *
    */
  override def invoke(value: String, context: SinkFunction.Context[_]): Unit = {


    val stat: PreparedStatement = con.prepareStatement("insert into student(id,name,age,gender,clazz) values(?,?,?,?,?)")

    val split: Array[String] = value.split(",")
    val id: Int = split(0).toInt
    val name: String = split(1)
    val age: Int = split(2).toInt
    val gender: String = split(3)
    val clazz: String = split(4)


    stat.setInt(1, id)
    stat.setString(2, name)
    stat.setInt(3, age)
    stat.setString(4, gender)
    stat.setString(5, clazz)


    stat.executeUpdate()


  }

}
